﻿// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. 
// See License.txt in the project root for license information.
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Diagnostics.Contracts;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading;
using System.Threading.Actors;
using System.Threading.Tasks;

namespace System.Cloud.Collections
{
    // Represents an IDictionary<String, TValue> stored in an actor, in the cloud.
    // Note that this class is prone to all the race conditions inherent with ConcurrentDictionary.
    public class CloudStringDictionary<TValue> : IDictionaryAsync<String, TValue>, IDictionary<String, TValue>, IDisposable
    {
        internal ActorClient _actorClient;
        private String _dictionaryName;
        private String _partitionName;
        private bool _deleteActor;

        private const String ApplicationType = "fabric:/actor/dictionary";

        private readonly Object[] DictionaryNameOneElementArray;

        public CloudStringDictionary(String hostEndpoint, String dictionaryName, /*CancellationToken cancellationToken,*/ 
            bool connectThroughGateway = false, int numReplicas = 2)
            : this(hostEndpoint, dictionaryName, typeof(String), connectThroughGateway, numReplicas)
        {
            Contract.Requires(!String.IsNullOrEmpty(hostEndpoint));
            Contract.Requires(!String.IsNullOrEmpty(dictionaryName));            
        }

        // This constructor is to support CloudDictionary<TKey, TValue>.  Pass TKey for the keyConstraintType.
        // For all uses of CloudStringDictionary, obviously keyConstraintType is typeof(String).
        internal CloudStringDictionary(String hostEndpoint, String dictionaryName, Type keyConstraintType, 
            /*CancellationToken cancellationToken,*/ bool connectThroughGateway, int numReplicas)
        {
            Contract.Requires(!String.IsNullOrEmpty(hostEndpoint));
            Contract.Requires(!String.IsNullOrEmpty(dictionaryName));
            Contract.Requires(keyConstraintType != null);

            Uri actorUri = new Uri(ApplicationType + "/" + dictionaryName);
            _dictionaryName = dictionaryName;
            DictionaryNameOneElementArray = new Object[] { _dictionaryName };

            // See if this collection exists.  If not, create it.
            _actorClient = ActorClient.CreateOrConnectToActor(hostEndpoint, ApplicationType, actorUri, 
                                                              "DictionaryActorReplica", connectThroughGateway, numReplicas);

            // Deploy TValue's assembly to the server, if necessary.
            // @TODO: Look at types these type parameters depends on.  IE, is TValue a Tuple<UserType1, UserType2>?
            // Update: .NET type information is no longer needed on the actor.  The actor stores pure JSON.
            //_actorClient.DeployAssemblyIfNeeded(typeof(TValue).GetTypeInfo().Assembly);

            _actorClient.CallMethod("TryConstruct", new Object[] { dictionaryName, keyConstraintType.FullName, typeof(TValue).FullName });
        }

        // Use built-in partitioning support, which is somewhat incompatible with a naive approach to connection pooling.
        public CloudStringDictionary(String hostEndpoint, String dictionaryName, String partitionName, 
            int numPartitions, /*CancellationToken cancellationToken,*/ bool connectThroughGateway, int numReplicas = 2)
            : this(hostEndpoint, dictionaryName, partitionName, typeof(String), numPartitions, connectThroughGateway, numReplicas)
        {
            Contract.Requires(!String.IsNullOrEmpty(hostEndpoint));
            Contract.Requires(!String.IsNullOrEmpty(partitionName));
            Contract.Requires(!String.IsNullOrEmpty(dictionaryName));
            Contract.Requires(numPartitions >= 1, "Actors must have one or more partitions.");
        }
        
        internal CloudStringDictionary(String hostEndpoint, String dictionaryName, String partitionName, Type keyConstraintType, 
            int numPartitions, /*CancellationToken cancellationToken,*/ bool connectThroughGateway, int numReplicas = 2)
        {
            Contract.Requires(!String.IsNullOrEmpty(hostEndpoint));
            Contract.Requires(!String.IsNullOrEmpty(dictionaryName));
            Contract.Requires(!String.IsNullOrEmpty(partitionName));
            Contract.Requires(numPartitions >= 1, "Actors must have one or more partitions.");
            Contract.Requires(keyConstraintType != null);

            Uri actorUri = new Uri(ApplicationType + "/" + dictionaryName);
            _dictionaryName = partitionName;
            DictionaryNameOneElementArray = new Object[] { _dictionaryName };
            _partitionName = partitionName;

            // See if this collection exists.  If not, create it.
            _actorClient = ActorClient.CreateOrConnectToActor(hostEndpoint, ApplicationType, actorUri,
                                                              "DictionaryActorReplica", connectThroughGateway, numReplicas, numPartitions, _partitionName);

            // Deploy TValue's assembly to the server, if necessary.
            // @TODO: Look at types these type parameters depends on.  IE, is TValue a Tuple<UserType1, UserType2>?
            // Update: .NET type information is no longer needed on the actor.  The actor stores pure JSON.
            //_actorClient.DeployAssemblyIfNeeded(typeof(TValue).GetTypeInfo().Assembly);

            _actorClient.CallMethod("TryConstruct", new Object[] { partitionName, keyConstraintType.FullName, typeof(TValue).FullName });
        }


        #region Lifetime management

        ~CloudStringDictionary()
        {
            // Ensure we close the ActorClient.
            Dispose(false);
        }

        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        protected virtual void Dispose(bool disposing)
        {
            if (_actorClient != null)
            {
                if (disposing)
                {
                    if (_deleteActor)
                    {
                        // @TODO: I have not heard that there is a way to delete one single partitioned data structure within a
                        // partition, only the ability to delete the entire service instance.  Figure this out with our partners.
                        if (_partitionName == null)
                            _actorClient.DeleteActor();
                    }
                }

                // Note that we're calling Dispose() here on _actorClient, which may have theoretically already been 
                // finalized if disposing is false.  However, if someone creates a CloudList<T> and leaks it, we need 
                // some way to close the connection - remember we're doing an async read, so the continuation itself which isn't
                // even running is sitting on that socket and holding the ActorClient instance live.  Calling Dispose
                // on it will force the connection to close, and get rid of that continuation.
                _actorClient.Dispose();

                _actorClient = null;
            }
        }

        // Destroys the dictionary stored in the Cloud
        public void Destroy()
        {
            // @TODO: Revisit this once we talk about co-locating collections on a single actor.  Destroying the dictionary != deleting the actor.
            _actorClient.CallMethod("Destroy", DictionaryNameOneElementArray);
            String fabricAddress = _actorClient.ServerUri.Host + ":" + _actorClient.ServerUri.Port;
            _deleteActor = true;
            Dispose();
        }

        #endregion Lifetime management

        #region IAsyncDictionary<String, TValue> members

        // @TODO: Not yet plumbed through to FabricActorClient yet.
        public TimeSpan Timeout
        {
            get;
            set;
        }

        public Task<TValue> GetValueAsync(String key)
        {
            if (key == null)
                throw new ArgumentNullException("key");
            Contract.EndContractBlock();

            return _actorClient.CallMethodAsync<TValue>("Get", new Object[] { _dictionaryName, key });
        }

        public Task SetValueAsync(String key, TValue value)
        {
            if (key == null)
                throw new ArgumentNullException("key");
            Contract.EndContractBlock();

            return _actorClient.CallMethodAsync<int>("Set", new Object[] { _dictionaryName, key, value });
        }

        public Task<Tuple<bool, TValue>> TryGetValueAsync(String key)
        {
            if (key == null)
                throw new ArgumentNullException("key");
            Contract.EndContractBlock();

            // Value types may return null - we must carefully deserialize them.  But normal reference types
            // like a List<T> are best deserialized by passing the precise type from the client to the 
            // deserializer (by passing the type to CallMethodAsync).
            Task<Tuple<bool, TValue>> ret;
            if (default(TValue) != null)
            {
                ret = _actorClient.CallMethodAsync<Tuple<bool, Object>>("TryGetValue", new Object[] { _dictionaryName, key }).ContinueWith((completed) =>
                {
                    Tuple<bool, Object> tuple = completed.Result;
                    TValue value = (tuple.Item2 != null) ? ConvertToTValue(tuple.Item2) : default(TValue);
                    return new Tuple<bool, TValue>(tuple.Item1, value);
                });
            }
            else
            {
                ret = _actorClient.CallMethodAsync<Tuple<bool, TValue>>("TryGetValue", new Object[] { _dictionaryName, key });
            }
            return ret;
        }

        public Task<ICollection<TValue>> GetValuesAsync(IEnumerable<String> keys)
        {
            if (keys == null)
                throw new ArgumentNullException("keys");
            Contract.EndContractBlock();

            Object[] parameters = FlattenParameters(keys);
            return _actorClient.CallMethodAsync<ICollection<TValue>>("MultiGet", parameters);
        }

        public Task SetValuesAsync(IEnumerable<String> keys, IEnumerable<TValue> values)
        {
            if (keys == null)
                throw new ArgumentNullException("keys");
            if (values == null)
                throw new ArgumentNullException("values");
            Contract.EndContractBlock();

            // Flatten parameters
            Object[] parameters = FlattenParameters(keys, values);
            return _actorClient.CallMethodAsync<int>("Set", parameters);
        }

        public Task<bool> ContainsKeyAsync(String key)
        {
            if (key == null)
                throw new ArgumentNullException("key");
            Contract.EndContractBlock();

            return _actorClient.CallMethodAsync<bool>("ContainsKey", new Object[] { _dictionaryName, key });
        }

        public async Task<bool> RemoveAsync(String key)
        {
            if (key == null)
                throw new ArgumentNullException("key");
            Contract.EndContractBlock();

            int numRemoved = await _actorClient.CallMethodAsync<int>("Remove", new Object[] { _dictionaryName, key }).ConfigureAwait(continueOnCapturedContext:false);
            return numRemoved != 0;
        }

        public Task RemoveAsync(IEnumerable<String> keys)
        {
            if (keys == null)
                throw new ArgumentNullException("keys");
            Contract.EndContractBlock();

            Object[] parameters = FlattenParameters(keys);
            return _actorClient.CallMethodAsync<int>("Remove", parameters);
        }

        public ICollection<String> Keys
        {
            get { return _actorClient.CallMethod<ICollection<String>>("EnumKeys", new Object[] { _dictionaryName }); }
        }

        public ICollection<TValue> Values
        {
            get { return _actorClient.CallMethod<ICollection<TValue>>("EnumValues", new Object[] { _dictionaryName }); }
        }

        public Task<int> CountAsync
        {
            get { return _actorClient.CallMethodAsync<int>("Count", DictionaryNameOneElementArray); }
        }

        public Task<bool> IsReadOnlyAsync
        {
            get { return _actorClient.CallMethodAsync<bool>("IsReadOnly", DictionaryNameOneElementArray); }
        }

        public Task AddAsync(KeyValuePair<String, TValue> item)
        {
            if (item.Key == null)
                throw new ArgumentNullException("item.Key");
            Contract.EndContractBlock();

            return _actorClient.CallMethodAsync<int>("Set", new Object[] { _dictionaryName, item.Key, item.Value });
        }

        public Task ClearAsync()
        {
            return _actorClient.CallMethodAsyncVoid("Clear", DictionaryNameOneElementArray);
        }

        public Task<bool> ContainsAsync(KeyValuePair<String, TValue> item)
        {
            if (item.Key == null)
                throw new ArgumentNullException("item.Key");
            Contract.EndContractBlock();

            return ContainsKeyAsync(item.Key);
        }

        public Task CopyToAsync(KeyValuePair<String, TValue>[] array, int arrayIndex)
        {
            if (array == null)
                throw new ArgumentNullException("array");
            if (arrayIndex < 0 || arrayIndex >= array.Length)
                throw new ArgumentOutOfRangeException("arrayIndex");
            Contract.EndContractBlock();

            return _actorClient.CallMethodAsync<IList<KeyValuePair<String, TValue>>>("Enumerate", DictionaryNameOneElementArray)
                .ContinueWith(completedTask =>
                {
                    if (!completedTask.IsFaulted)
                        completedTask.Result.CopyTo(array, arrayIndex);
                });
        }

        Task<bool> ICollectionAsync<KeyValuePair<String, TValue>>.RemoveAsync(KeyValuePair<String, TValue> item)
        {
            if (item.Key == null)
                throw new ArgumentNullException("item.Key");
            Contract.EndContractBlock();

            return RemoveAsync(item.Key);
        }

        public IDisposable Subscribe(IObserver<KeyValuePair<String, TValue>> observer)
        {
            throw new NotImplementedException();
        }

        #endregion IAsyncDictionary<String, TValue> members

        #region IDictionary<String, TValue> members

        public void Add(String key, TValue value)
        {
            if (key == null)
                throw new ArgumentNullException("key");
            Contract.EndContractBlock();

            _actorClient.CallMethod("Set", new Object[] { _dictionaryName, key, value });
        }

        public bool ContainsKey(String key)
        {
            if (key == null)
                throw new ArgumentNullException("key");
            Contract.EndContractBlock();

            return _actorClient.CallMethod<bool>("ContainsKey", new Object[] { _dictionaryName, key });
        }

        ICollection<String> IDictionary<String, TValue>.Keys
        {
            get { return _actorClient.CallMethod<ICollection<String>>("EnumKeys", new Object[] { _dictionaryName }); }
        }

        public bool TryGetValue(String key, out TValue value)
        {
            if (key == null)
                throw new ArgumentNullException("key");
            Contract.EndContractBlock();

            // Value types may return null - we must carefully deserialize them.  But normal reference types
            // like a Lis<T> are best deserialized by passing the precise type from the client to the 
            // deserializer (by passing the type to CallMethod).
            if (default(TValue) != null)
            {
                var objectTuple = _actorClient.CallMethod<Tuple<bool, Object>>("TryGetValue", new Object[] { _dictionaryName, key });
                value = (objectTuple.Item2 != null) ? ConvertToTValue(objectTuple.Item2) : default(TValue);
                return objectTuple.Item1;
            }
            else
            {
                var ret = _actorClient.CallMethod<Tuple<bool, TValue>>("TryGetValue", new Object[] { _dictionaryName, key });
                value = (ret.Item2 != null) ? ret.Item2 : default(TValue);
                return ret.Item1;
            }
        }

        // Work around for two problems.  First, this is necessary because we return a 
        // Tuple<String, Object> which can have a null in it, but we can't directly deserialize that to a TValue
        // if TValue is a value type.  
        // Second, work around JSON's number representation problems.  
        private TValue ConvertToTValue(Object obj)
        {
            if (typeof(TValue) == typeof(Single))
                return (TValue)(Object)(float)(double)obj;

            return (TValue)obj;
        }

        ICollection<TValue> IDictionary<String, TValue>.Values
        {
            get { return _actorClient.CallMethod<ICollection<TValue>>("EnumValues", new Object[] { _dictionaryName }); }
        }

        public TValue this[String key]
        {
            get
            {
                if (key == null)
                    throw new ArgumentNullException("key");
                Contract.EndContractBlock();

                return _actorClient.CallMethod<TValue>("Get", new Object[] { _dictionaryName, key });
            }
            set
            {
                if (key == null)
                    throw new ArgumentNullException("key");
                Contract.EndContractBlock();

                _actorClient.CallMethod("Set", new Object[] { _dictionaryName, key, value });
            }
        }

        public void Add(KeyValuePair<String, TValue> item)
        {
            if (item.Key == null)
                throw new ArgumentNullException("item.Key");
            Contract.EndContractBlock();

            _actorClient.CallMethod("Add", new Object[] { _dictionaryName, item });
        }

        public void Clear()
        {
            _actorClient.CallMethod("Clear", DictionaryNameOneElementArray);
        }

        // This is a goofy method.  Privately implementing it so people generally don't call it.
        bool ICollection<KeyValuePair<String, TValue>>.Contains(KeyValuePair<String, TValue> item)
        {
            return ContainsKey(item.Key);
        }

        public void CopyTo(KeyValuePair<String, TValue>[] array, int arrayIndex)
        {
            if (array == null)
                throw new ArgumentNullException("array");
            if (arrayIndex < 0 || arrayIndex >= array.Length)
                throw new ArgumentOutOfRangeException("arrayIndex");
            Contract.EndContractBlock();

            var pairs = _actorClient.CallMethod<IList<KeyValuePair<String, TValue>>>("Enumerate", DictionaryNameOneElementArray);
            pairs.CopyTo(array, arrayIndex);
        }

        public int Count
        {
            get { return _actorClient.CallMethod<int>("Count", DictionaryNameOneElementArray); }
        }

        public bool IsReadOnly
        {
            get { return _actorClient.CallMethod<bool>("IsReadOnly", DictionaryNameOneElementArray); }
        }

        public bool Remove(String key)
        {
            if (key == null)
                throw new ArgumentNullException("key");
            Contract.EndContractBlock();

            return _actorClient.CallMethod<bool>("Remove", new Object[] { _dictionaryName, key });
        }

        // Silly method
        bool ICollection<KeyValuePair<String, TValue>>.Remove(KeyValuePair<String, TValue> item)
        {
            return Remove(item.Key);
        }

        public IEnumerator<KeyValuePair<String, TValue>> GetEnumerator()
        {
            var pairs = _actorClient.CallMethod<IList<KeyValuePair<String, TValue>>>("Enumerate", DictionaryNameOneElementArray);
            return pairs.GetEnumerator();
        }

        System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator()
        {
            return GetEnumerator();
        }

        #endregion IDictionary<String, TValue> members

        #region Helper methods

        private Object[] FlattenParameters(IEnumerable<String> keys)
        {
            Contract.Requires(keys != null);
            Contract.Ensures(Contract.Result<Object[]>() != null);

            String[] keyArray = keys.ToArray();
            Object[] parameters = new Object[1 + keyArray.Length];
            parameters[0] = _dictionaryName;
            Array.Copy(keyArray, 0, parameters, 1, keyArray.Length);
            return parameters;
        }

        private Object[] FlattenParameters(IEnumerable<String> keys, IEnumerable<TValue> values)
        {
            Contract.Requires(keys != null);
            Contract.Requires(values != null);
            Contract.Requires(keys.Count() == values.Count());
            Contract.Ensures(Contract.Result<Object[]>() != null);

            String[] keyArray = keys.ToArray();
            TValue[] valueArray = values.ToArray();
            Object[] parameters = new Object[1 + 2 * keyArray.Length];
            parameters[0] = _dictionaryName;
            int index = 1;
            for (int i = 0; i < keyArray.Length; i++)
            {
                parameters[index++] = keyArray[i];
                parameters[index++] = valueArray[i];
            }
            return parameters;
        }

        #endregion Helper methods
    }
}
